package com.tensai.study.java.newconcept.typereactiveprogram.reactor.schedule;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author Tensai
 */
@Slf4j
public class SchedulersTest {

	/**
	 * 提交的Runnable将会立马在当前线程执行
	 */
	@Test
	public void immediateTest() {
		schedule(Schedulers.immediate());
	}

	/**
	 * 使用同一个子线程来执行所有的任务
	 */
	@Test
	public void singleTest() {
		schedule(Schedulers.single());
	}

	/**
	 * 创建一个可重用的线程池，如果线程池中的线程在长时间内都没有被使用，那么将会被回收。
	 * boundedElastic会有一个最大的线程个数，一般来说是CPU cores x 10。
	 * 如果目前没有可用的worker线程，提交的任务将会被放入队列等待
	 */
	@Test
	public void boundedElasticTest() {
		schedule(Schedulers.boundedElastic());
	}

	/**
	 * 定时任务调度 固定时间间隔
	 */
	@Test
	public void schedulePeriodicallyTest() throws InterruptedException {
		Scheduler scheduler = Schedulers.boundedElastic();
		Disposable defaultDisposable = scheduler.schedulePeriodically(getRunnable("task--"), 5, 4, TimeUnit.SECONDS);
		Map<Integer, Disposable> disposableMap = new HashMap<>(32);
		disposableMap.put(-1, defaultDisposable);
		for (int i = 0; i < 10; i++) {
			Disposable disposable = scheduler.schedulePeriodically(getRunnable("task" + i), 3, i + 5, TimeUnit.SECONDS);
			disposableMap.put(i, disposable);
		}
		Thread.sleep(20000);
		disposableMap.get(9).dispose();
		for (; ; ) {
		}
	}

	private void schedule(Scheduler scheduler) {
		log.info("start");
		for (int i = 0; i < 100; i++) {
			scheduler.schedule(TestData.log("立即执行"));
			log.info("{}", scheduler.now(TimeUnit.SECONDS));
		}
	}

	private Runnable getRunnable(String msg) {
		return () -> log.info(msg);
	}
}
